/*
 * Copyright 2019-2025 the original author or authors.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * https://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.iiifi.kite.rocketmq.configuration;

import java.util.Map;
import java.util.concurrent.*;

import javax.annotation.PostConstruct;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import com.iiifi.kite.rocketmq.annotation.RocketMqProducer;
import com.iiifi.kite.rocketmq.annotation.RocketMqTransactionProducer;
import com.iiifi.kite.rocketmq.core.AbstractMqTransactionProducer;

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * 自动装配消息生产者
 *
 * @author kite@iiifi.com 花朝
 */
@Slf4j
@Configuration
@ConditionalOnBean(RocketMqAutoConfiguration.class)
public class RocketMqProducerAutoConfiguration extends RocketMqAutoConfiguration {

    @Setter
    private static DefaultMQProducer producer;

    @Bean
    public DefaultMQProducer exposeProducer() throws Exception {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RocketMqProducer.class);
        // 对于仅仅只存在消息消费者的项目，无需构建生产者
        if (CollectionUtils.isEmpty(beans)) {
            return null;
        }
        if (producer == null) {
            Assert.notNull(mqProperties.getProducerGroup(), "producer group must be defined");
            Assert.notNull(mqProperties.getNameServerAddress(), "name server address must be defined");
            producer = new DefaultMQProducer(mqProperties.getProducerGroup());
            producer.setNamesrvAddr(mqProperties.getNameServerAddress());
            producer.setSendMsgTimeout(mqProperties.getSendMsgTimeout());
            producer.setSendMessageWithVIPChannel(mqProperties.getVipChannelEnabled());
            producer.start();
        }
        return producer;
    }

    @PostConstruct
    public void configTransactionProducer() {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RocketMqTransactionProducer.class);
        if (CollectionUtils.isEmpty(beans)) {
            return;
        }
        ExecutorService executorService = new ThreadPoolExecutor(beans.size(), beans.size() * 2, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable runnable) {
                        Thread thread = new Thread(runnable);
                        thread.setName("client-transaction-msg-check-thread");
                        return thread;
                    }
                });
        Environment environment = applicationContext.getEnvironment();
        beans.forEach((key, value) -> {
            try {
                AbstractMqTransactionProducer beanObj = (AbstractMqTransactionProducer) value;
                RocketMqTransactionProducer rocketMqTransactionProducer =
                        beanObj.getClass().getAnnotation(RocketMqTransactionProducer.class);
                TransactionMQProducer producer = new TransactionMQProducer(
                        environment.resolvePlaceholders(rocketMqTransactionProducer.producerGroup()));
                producer.setNamesrvAddr(mqProperties.getNameServerAddress());
                producer.setExecutorService(executorService);
                producer.setTransactionListener(beanObj);
                producer.start();
                beanObj.setProducer(producer);
            } catch (Exception e) {
                log.error("build transaction producer error : {}", e.getMessage());
            }
        });
    }
}
